package com.lndsnx.demo;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.lndsnx.config.AmqpConfig;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

/**
 * @Description 消费者
 * @Author lndsnx
 * @Date 2025/7/30 上午9:50
 **/
public class Consumer {
    // 设置Queue名称。需要和生产者中设置的Queue名称一致。
    private static final String QUEUE_NAME = "queue-test";
    // 虚拟主机名称。需要和生产者中设置的虚拟主机名称一致。
    private static final String VHOST = "vhost_test";
    // 消费者标签。
    private static final String CONSUMER_TAG = "tag_test";
    // 设置RabbitMQ实例的接入点。
    private static final String HOSTNAME = AmqpConfig.HOSTNAME;
    // 账号。需要和生产者中设置的账号一致。
    private static final String USERNAME = AmqpConfig.USERNAME;
    // 密码。需要和生产者中设置的密码一致。
    private static final String PASSWORD = AmqpConfig.PASSWORD;

    //如果需要使用 TLS 加密，改为 5671 端口，并且enableSSL设置为true。。
    public static final int PORT = 5672;
    public static final boolean enableSSL = false;

    // 创建连接通道。
    private final Channel channel;
    // 设置Queue名称。需要和生产者中设置的Queue名称一致。
    private final String queue;

    /**
     * 创建消费者。初始化通道和队列名称
     * @param channel 通道
     * @param queue 队列
     */
    public Consumer(Channel channel, String queue) {
        this.channel = channel;
        this.queue = queue;
    }

    public static void main(String[] args) throws Exception {
        // 创建连接工厂。
        ConnectionFactory factory = new ConnectionFactory(HOSTNAME, PORT, USERNAME, PASSWORD, VHOST, enableSSL);
        // 创建通道。
        Channel channel = factory.createChannel();
        // 设置qos。控制未确认消息的数量
        channel.basicQos(50);


        //设置为云消息队列 RabbitMQ 版实例的Queue名称。需要和生产者中设置的Queue名称一致。
        Consumer consumer = new Consumer(channel, QUEUE_NAME);

        // 创建消费者。
        consumer.consume();
    }

    public void consume() throws Exception {
        // 启动消费者并监听队列消息
        channel.basicConsume(queue, false, CONSUMER_TAG, new DefaultConsumer(channel) {
            /**
             * 接收到消息时调用。
             * @param consumerTag 消费者标签。
             * @param envelope
             * @param properties 消息属性。
             * @param body 消息内容。
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {

                // 业务处理。
                // 打印接收到的消息ID
                System.out.println("receive: msgId=" + properties.getMessageId());

                // 消费者需要在有效时间内提交ack，否则消息会重新推送，最多推送16次。
                // 若推送16次还未成功，则消息被丢弃或者进入死信Exchange。
                // 专业版实例的有效时间为1分钟，企业版和Serverless实例为5分钟，铂金版实例为30分钟。
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });

        // 创建倒计时锁存器用于优雅关闭
        // 向Java虚拟机注册了一个关闭钩子（shutdown hook）。
        // 当JVM接收到终止信号（如通过系统中断或System.exit()调用）时，会执行这个钩子中的代码,关闭connection,退出程序
        CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                channel.getConnection().close();
            } catch (IOException e) {
                System.out.println("close connection error." + e);
            }
            latch.countDown();
        }));
        latch.await();
    }
}

